package bjut.zrpc.core.optimization.netty;


import bjut.zrpc.core.optimization.channel.ChannelRecord;
import bjut.zrpc.core.optimization.constant.ZRpcConstant;
import bjut.zrpc.core.optimization.enums.ZRpcCompressEnum;
import bjut.zrpc.core.optimization.enums.ZRpcConfigEnum;
import bjut.zrpc.core.optimization.enums.ZRpcSerializerEnum;
import bjut.zrpc.core.optimization.factory.SingletonFactory;
import bjut.zrpc.core.optimization.interfaces.ZRpcRequestTransport;
import bjut.zrpc.core.optimization.message.ZRpcMessage;
import bjut.zrpc.core.optimization.message.ZRpcRequestMessage;
import bjut.zrpc.core.optimization.message.ZRpcResponseMessage;
import bjut.zrpc.core.optimization.processor.UnprocessedRequest;
import bjut.zrpc.core.optimization.utils.PropertiesReadUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @Author: zhuzhiming
 * @Date: 2022/6/24 14:36
 */
@Slf4j
public class NettyClient implements ZRpcRequestTransport {

    private final EventLoopGroup workerGroup;
    private final Bootstrap bootstrap;
    private final ChannelRecord channelRecord;

    private final UnprocessedRequest unprocessedRequest;



    public NettyClient() {
        workerGroup = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(workerGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .handler(new LoggingHandler(LogLevel.INFO))
                .handler(new NettyClientInitializer());

        channelRecord = SingletonFactory.getInstance(ChannelRecord.class);
        unprocessedRequest = SingletonFactory.getInstance(UnprocessedRequest.class);
    }


    @Override
    public Object sendRpcRequest(ZRpcRequestMessage zRpcRequestMessage, InetSocketAddress inetSocketAddress) {

        CompletableFuture<ZRpcResponseMessage<Object>> resultFuture = new CompletableFuture<>();

        Channel channel = getChannel(inetSocketAddress);
        if (channel.isActive()) {
            unprocessedRequest.putUnprocessedRequest(zRpcRequestMessage.getRequestId(), resultFuture);

            String compress = PropertiesReadUtil.getConfig(ZRpcConfigEnum.ZRPC_COMPRESS.getPropertyValue()[0]);
            String serializer = PropertiesReadUtil.getConfig(ZRpcConfigEnum.ZRPC_SERIALIZER.getPropertyValue()[0]);

            ZRpcMessage zRpcMessage = ZRpcMessage.builder().messageType(ZRpcConstant.REQUEST_TYPE)
                    .compress(compress.equals("true") ? ZRpcCompressEnum.GZIP.getType() : ZRpcCompressEnum.getCompressType(compress))
                    .codec(serializer.equals("true") ? ZRpcSerializerEnum.HESSIAN.getCodec() : ZRpcSerializerEnum.getSerializationType(serializer))
                    .bodyData(zRpcRequestMessage)
                    .build();

            channel.writeAndFlush(zRpcMessage).addListener((ChannelFutureListener) future -> {

                if (future.isSuccess()) {
                    log.info("client send message [{}] success", zRpcMessage);
                } else {
                    future.channel().close();
                    resultFuture.completeExceptionally(future.cause());
                    log.error("client send message [{}]  fail", zRpcMessage);
                }
            });
        }

        return resultFuture;

    }

    public Channel getChannel(InetSocketAddress inetSocketAddress) {

        Channel channel = channelRecord.getChannel(inetSocketAddress.toString());

        if (channel == null) {
            channel = doConnectServer(inetSocketAddress);
            channelRecord.putChannel(inetSocketAddress.toString(), channel);
        }

        return channel;
    }

    private Channel doConnectServer(InetSocketAddress inetSocketAddress) {

        CompletableFuture<Channel> completableFuture = new CompletableFuture<>();

        bootstrap.connect(inetSocketAddress).addListener((ChannelFutureListener) future -> {

            if (future.isSuccess()) {
                completableFuture.complete(future.channel());
            } else {
                completableFuture.completeExceptionally(future.cause());
            }
        });

        try {
            return completableFuture.get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e.getMessage());
        }
    }


}
